package com.xzx.flink.streamapi.watermark;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 断点式的水位线
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/4/30 13:58
 */
public class Watermark_Punctuated {
    private static final Logger log = LoggerFactory.getLogger(Watermark_Punctuated.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(new CustomPunctuatedWatermarkStrategy())
                .print();

        env.execute(Watermark_Punctuated.class.getSimpleName());
    }

    public static class CustomPunctuatedWatermarkStrategy implements WatermarkStrategy<ClickEvent> {

        // 时间戳分配器
        @Override
        public TimestampAssigner<ClickEvent> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<ClickEvent>() {
                @Override
                public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                    return element.getTimestamp();
                }
            };
        }

        // 水位线生成器
        @Override
        public WatermarkGenerator<ClickEvent> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new PunctuatedWatermarkGenerator();
        }
    }

    public static class PunctuatedWatermarkGenerator implements WatermarkGenerator<ClickEvent> {
        private static final Long DELAY = 2000L;// 延迟2秒
        //某个条件触发水位线
        @Override
        public void onEvent(ClickEvent event, long eventTimestamp, WatermarkOutput output) {
            log.info("=============onEvent====");
            if (event.getUser().equals("Mary")) {
                log.info("=============emitWatermark====");
                output.emitWatermark(new Watermark(eventTimestamp - DELAY - 1));
            }
        }

        // 什么都不用做
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {

        }
    }
}
